package org.zjt.flink.stream;

import lombok.extern.slf4j.Slf4j;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.util.Random;


/**
 * Description:
 *
 * @author juntao.zhang
 * Date: 2018-10-11 下午9:07
 * @see
 */
public class WindowSinkJob {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment executionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();

        Random random = new Random();

        executionEnvironment.addSource(new SourceFunction<Integer>() {
            @Override
            public void run(SourceContext<Integer> ctx) throws Exception {
                for (int i = 0; i < 1000; i++) {
                    ctx.collect(random.nextInt(1000));
                }
            }

            @Override
            public void cancel() {
            }
        })
                .addSink(new MySinkFunction());

        executionEnvironment.execute();
    }

    @Slf4j
    public static class MySinkFunction implements SinkFunction {
        @Override
        public void invoke(Object value, Context context) throws Exception {
            log.info("发送器 invoke :{}",value);
        }
    }
}
